package cn.demoncat.util.lang;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.function.Supplier;

/**
 * 异步工具
 *
 * 1、默认使用ForkJoinPool.commonPool()，执行CompletableFuture异步操作
 *
 * whenComplete()：执行完成（成功或异常）回调（由任务线程执行）
 * whenCompleteAsync()：执行完成（成功或异常）回调（由新线程执行）
 * exceptionally()：发生异常回调，可以修改返回值
 * handle()：添加串行任务（成功或异常），返回结果
 * thenApply()：添加串行任务（成功），返回结果
 * thenAccept()：串行执行任务，有消费，无返回
 * thenRun()：串行执行任务，无消费，不返回
 * get()：阻塞等待执行完成，抛出ExecutionException
 * join()：阻塞等待执行完成，抛出CompletionException（RuntimeException）
 *
 * 执行完成：whenComplete()
 *
 * @author 延晓磊
 * @since 2021年02月18日
 */
public class CompletableFutureUtil {

    /**
     * 所有任务完成时，返回新的CompletableFuture（异步执行，阻塞等待全部完成）
     *
     * @param cfs   异步任务
     *
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
        return CompletableFuture.allOf(cfs);
    }

    /**
     * 任意任务完成时，返回新的CompletableFuture（异步执行，阻塞等待任意完成）
     *
     * @param cfs   异步任务
     *
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
        return CompletableFuture.anyOf(cfs);
    }

    /**
     * 阻塞等待任务执行结束
     *
     * @param future
     * @return 执行结果
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static <T>T get(CompletableFuture<T> future){
        try {
            return future.get();
        } catch (InterruptedException e) {
            // 执行中断
            throw new RuntimeException("异步任务执行中断", e);
        } catch (ExecutionException e) {
            // 执行失败
            throw new RuntimeException("异步任务执行失败", e);
        }
    }

    /**
     * 阻塞等待所有任务执行结束
     *
     * @param cfs
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static void done(CompletableFuture<?>... cfs) {
        get(allOf(cfs));
    }

    /**
     * 创建异步执行任务（异步执行，非阻塞）
     *
     * 注：使用 ForkJoinPool.commonPool()
     *
     * @param runnable  无返回的任务
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static CompletableFuture<Void> build(Runnable runnable) {
        return CompletableFuture.runAsync(runnable);
    }

    /**
     * 创建异步执行任务（异步执行，非阻塞）
     *
     *
     * @param runnable  无返回的任务
     * @param executor  任务执行器
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static CompletableFuture<Void> build(Runnable runnable, Executor executor) {
        return CompletableFuture.runAsync(runnable, executor);
    }

    /**
     * 创建异步执行任务（异步执行，非阻塞）
     *
     * 注：使用 ForkJoinPool.commonPool()
     *
     * @param supplier  有返回的任务
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static <U> CompletableFuture<U> build(Supplier<U> supplier) {
        return CompletableFuture.supplyAsync(supplier);
    }

    /**
     * 创建异步执行任务（异步执行，非阻塞）
     *
     * @param supplier  有返回的任务
     * @param executor  任务执行器
     * @return
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static <U> CompletableFuture<U> build(Supplier<U> supplier, Executor executor) {
        return CompletableFuture.supplyAsync(supplier, executor);
    }

    /**
     * 批量异步执行（异步执行，阻塞等待全部完成）
     *
     * 注：使用 ForkJoinPool.commonPool()
     *
     * @param runnable
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static void async(Runnable ...runnable){
        done(Arrays.stream(runnable).map(CompletableFutureUtil::build).toArray(CompletableFuture[]::new));
    }

    /**
     * 批量异步执行（异步执行，阻塞等待全部完成）
     *
     * 注：使用 ForkJoinPool.commonPool()
     *
     * @param runnable
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static void async(List<Runnable> runnable){
        done(runnable.stream().map(CompletableFutureUtil::build).toArray(CompletableFuture[]::new));
    }

    /**
     * 批量异步执行（异步执行，阻塞等待全部完成）
     *
     * @param executor  任务执行器
     * @param runnable
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static void async(Executor executor, Runnable ...runnable){
        done(Arrays.stream(runnable).map(r -> build(r, executor)).toArray(CompletableFuture[]::new));
    }

    /**
     * 批量异步执行（异步执行，阻塞等待全部完成）
     *
     * @param executor  任务执行器
     * @param runnable
     *
     * @author 延晓磊
     * @since 2021年02月18日
     */
    public static void async(Executor executor, List<Runnable> runnable){
        done(runnable.stream().map(r -> build(r, executor)).toArray(CompletableFuture[]::new));
    }
}
